package org.pr.ddddesigner.infrastructure.helper;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.annotation.Id;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.query.Query;
import org.springframework.data.relational.core.query.Update;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.DatabaseClient.GenericExecuteSpec;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.beans.BeanInfo;
import java.beans.IntrospectionException;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.io.Serializable;
import java.lang.invoke.SerializedLambda;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;

@Slf4j
public final class R2dbcHelper {
    private final static String THROWABLE_UNIQUE_CONSTRAINT = "duplicate key value violates unique constraint";

    private final static Map<Class<?>, TableInfo<?>> TABLE_INFO_CACHE = new ConcurrentHashMap<>();

    private final static Function<?, ?> IDENTITY = (Function<Object, Object>) o -> o;

    @FunctionalInterface
    public interface FieldFn<T, R> extends Function<T, R>, Serializable {
    }

    private static final Map<FieldFn<?, ?>, String> FIELD_FN_CACHE = new ConcurrentHashMap<>();

    private R2dbcHelper() throws IllegalAccessException {
        throw new IllegalAccessException();
    }

    public static boolean isUniqueException(Throwable e) {
        return e.getMessage().contains(THROWABLE_UNIQUE_CONSTRAINT);
    }

    public static Update update(String column, Object value) {
        return Update.update(column, value);
    }

    public static <T> Update update(FieldFn<T, ?> column, Object value) {
        return Update.update(toFieldName(column), value);
    }

    public static Criteria like(String search, String... columns) {
        Criteria criteria = Criteria.empty();

        if (StringUtils.hasText(search) && columns.length > 0) {
            StringJoiner joiner = new StringJoiner("%", "%", "%");

            Arrays.stream(search.split("[ |+]]")).filter(StringUtils::hasText).forEach(joiner::add);

            if (joiner.length() > 0) {
                String cond = joiner.toString();

                for (String column : columns) {
                    criteria = criteria.or(column).like(cond);
                }
            }
        }

        return criteria;
    }

    @SafeVarargs
    public static <T> Criteria like(String search, FieldFn<T, ?>... columns) {
        String[] fields = new String[columns.length];
        for (int i = 0; i < columns.length; i++) {
            fields[i] = toFieldName(columns[i]);
        }
        return like(search, fields);
    }

    public static Criteria isNotTrue(String column) {
        return where(column).isNull().or(where(column).isFalse());
    }

    public static <T> Criteria isNotTrue(FieldFn<T, ?> column) {
        return isNotTrue(toFieldName(column));
    }

    public static <T> QuerySpec<T> from(Class<T> domainType) {
        return new SelectSpecImpl<>(domainType, domainType, Query.empty(), null);
    }

    public static Criteria.CriteriaStep where(String column) {
        return Criteria.where(column);
    }

    public static <T> Criteria.CriteriaStep where(FieldFn<T, ?> fieldFn) {
        return Criteria.where(toFieldName(fieldFn));
    }

    public static <T> Mono<Long> batchInsert(Collection<T> collection, Class<T> cls) {
        if (collection == null || collection.isEmpty()) {
            return Mono.just(0L);
        }
        final List<T> rowList = new ArrayList<>(collection);
        // TODO 数据量过大时，分批次插入

        return Mono.defer(() -> {
            DatabaseClient client = SpringContext.getBean(DatabaseClient.class);
            TableInfo<T> tableInfo = getTableInfo(cls);

            StringJoiner columns = new StringJoiner(",");
            StringJoiner allParameters = new StringJoiner(",");
            Map<String, Object> parameterMap = new LinkedHashMap<>();

            int rowLen = rowList.size();
            for (int i = 0; i < rowLen; i++) {
                T e = rowList.get(i);
                String rowKey = "r" + i;

                StringJoiner parameters = new StringJoiner(",", "(", ")");
                List<ColumnInfo> columnList = tableInfo.columns;
                int colLen = columnList.size();
                for (int j = 0; j < colLen; j++) {
                    ColumnInfo column = columnList.get(j);
                    if (i == 0) {
                        columns.add(column.name);
                    }

                    String cellKey = rowKey + "_c" + j;
                    parameters.add(":" + cellKey);

                    Method getter = column.property.getReadMethod();
                    Object value = null;
                    if (getter != null) {
                        try {
                            getter.setAccessible(true);
                            value = getter.invoke(e);
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    }
                    if (value == null) {
                        parameterMap.put(cellKey, column.property.getPropertyType());
                    } else {
                        parameterMap.put(cellKey, value);
                    }
                }
                allParameters.add(parameters.toString());
            }

            String sql = String.format("INSERT INTO %s (%s) VALUES %s", tableInfo.tableName, columns, allParameters);
            GenericExecuteSpec spec = client.sql(sql);
            for (Map.Entry<String, Object> entry : parameterMap.entrySet()) {
                Object value = entry.getValue();
                if (value instanceof Class<?>) {
                    spec = spec.bindNull(entry.getKey(), (Class<?>) value);
                } else {
                    spec = spec.bind(entry.getKey(), value);
                }
            }

            return spec.fetch().rowsUpdated();
        });
    }

    @SuppressWarnings("unchecked")
    public static <T> TableInfo<T> getTableInfo(Class<T> cls) {
        return (TableInfo<T>) TABLE_INFO_CACHE.computeIfAbsent(cls, c -> {
            String tableName = getTableName(cls);
            List<ColumnInfo> columnList = new ArrayList<>();
            PropertyDescriptor idProperty = null;

            try {
                BeanInfo beanInfo = Introspector.getBeanInfo(cls, Object.class);

                for (PropertyDescriptor pd : beanInfo.getPropertyDescriptors()) {
                    Field field = ReflectionUtils.findField(cls, pd.getName());
                    if (field != null) {
                        ColumnInfo columnInfo = getColumnInfo(pd, field);
                        columnList.add(columnInfo);
                        if (idProperty == null && columnInfo.isId()) {
                            idProperty = pd;
                        }
                    }
                }
            } catch (IntrospectionException e) {
                log.error("", e);
            }
            return new TableInfo<>(cls, tableName, idProperty, columnList);
        });
    }

    private static <T> String getTableName(Class<T> cls) {
        Table annotationTable = AnnotationUtils.findAnnotation(cls, Table.class);
        return annotationTable != null ? annotationTable.value() : cls.getSimpleName();
    }

    private static ColumnInfo getColumnInfo(PropertyDescriptor pd, Field field) {
        boolean isId = false;
        boolean insertable = true;
        boolean updatable = true;
        String name = pd.getName();
        {
            Id id = field.getAnnotation(Id.class);
            if (id != null) {
                isId = true;
            } /*else {
                javax.persistence.Id id2 = field.getAnnotation(javax.persistence.Id.class);
                isId = id2 != null;
            }*/
        }
        {
            Column column = field.getAnnotation(Column.class);
            if (column != null) {
                name = column.value();
            } /*else {
                javax.persistence.Column column2 = field.getAnnotation(javax.persistence.Column.class);
                if (column2 != null) {
                    insertable = column2.insertable();
                    updatable = column2.updatable();
                    name = column2.name();
                }
            }*/
        }
        return new ColumnInfo(isId, insertable, updatable, name, pd);
    }

    static <T> String toFieldName(FieldFn<T, ?> fieldFn) {
        String fieldName = FIELD_FN_CACHE.get(fieldFn);
        if (fieldName != null) {
            return fieldName;
        }

        Exception ex = null;
        try {
            SerializedLambda sLambda = ReflectHelper.invokeDeclaredMethod(fieldFn, "writeReplace");

            Class<?> capturingClass = ReflectHelper.getDeclaredFieldValue(sLambda, "capturingClass");
            ClassLoader classLoader = capturingClass.getClassLoader();

            Class<?> implClass = classLoader.loadClass(sLambda.getImplClass().replaceAll("/", "."));

            String methodName = sLambda.getImplMethodName();

            if (methodName.startsWith("get") && methodName.length() > 3) {
                fieldName = StringUtils.uncapitalize(methodName.substring(3));
            } else if (methodName.startsWith("is") && methodName.length() > 2) {
                fieldName = StringUtils.uncapitalize(methodName.substring(2));
            }

            if (fieldName != null) {
                Field field = ReflectionUtils.findField(implClass, fieldName);
                if (field != null) {
                    Column annotation = AnnotationUtils.getAnnotation(field, Column.class);
                    if (annotation != null) {
                        fieldName = annotation.value();
                    }
                }

                FIELD_FN_CACHE.put(fieldFn, fieldName);
                return fieldName;
            }
        } catch (ReflectiveOperationException e) {
            ex = e;
        }
        throw new IllegalStateException("无法获取字段名", ex);
    }

    @SuppressWarnings("unchecked")
    private static <T, R> Function<T, R> identity() {
        return (Function<T, R>) IDENTITY;
    }

    static class TableInfo<T> {
        @Getter
        private final Class<T> cls;
        @Getter
        private final String tableName;
        @Getter
        private final PropertyDescriptor idProperty;
        @Getter
        private final List<ColumnInfo> columns;

        public TableInfo(Class<T> cls, String tableName, PropertyDescriptor idProperty, List<ColumnInfo> columns) {
            this.cls = cls;
            this.tableName = tableName;
            this.idProperty = idProperty;
            this.columns = columns != null ? Collections.unmodifiableList(columns) : Collections.emptyList();
        }
    }

    static class ColumnInfo {
        @Getter
        private final boolean id;
        @Getter
        private final boolean insertable;
        @Getter
        private final boolean updatable;
        @Getter
        private final String name;
        @Getter
        private final PropertyDescriptor property;

        public ColumnInfo(boolean id, boolean insertable, boolean updatable, String name, PropertyDescriptor property) {
            this.id = id;
            this.insertable = insertable;
            this.updatable = updatable;
            this.name = name;
            this.property = property;
        }

        public Object getPropertyValue(Object obj) {
            Method getter = property.getReadMethod();
            Object value = null;
            if (getter != null) {
                try {
                    getter.setAccessible(true);
                    value = getter.invoke(obj);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            return value;
        }
    }

    public interface FromSpec<T> extends QuerySpec<T> {
        <R> QuerySpec<R> as(Class<R> resultType);
    }

    public interface QuerySpec<T> extends TerminatingSpec<T> {
        MappingSpec<T> matching(Query query);

        MappingSpec<T> matching(Criteria criteria);

        MappingSpec<T> matching(Criteria criteria, Pageable pageable);
    }

    public interface MappingSpec<T> extends TerminatingSpec<T> {
        <R> TerminatingSpec<R> mapTo(Function<T, R> mapping);
    }

    public interface TerminatingSpec<T> {
        Mono<Page<T>> execute();

        Mono<Page<T>> execute(R2dbcEntityTemplate template);
    }

    public interface SelectSpec<T> extends FromSpec<T>, QuerySpec<T>, MappingSpec<T> {
    }

    static class SelectSpecImpl<T> implements SelectSpec<T> {
        private final Class<?> domainType;
        private final Class<?> returnType;
        private final Query query;
        @Nullable
        private final Function<?, ?> mapping;

        public SelectSpecImpl(Class<?> domainType, Class<?> returnType, Query query, @Nullable Function<?, ?> mapping) {
            this.domainType = domainType;
            this.returnType = returnType;
            this.query = query;
            this.mapping = mapping;
        }

        @Override
        public <R> QuerySpec<R> as(Class<R> returnType) {
            return new SelectSpecImpl<>(domainType, returnType, query, null);
        }

        @Override
        public MappingSpec<T> matching(Query query) {
            return new SelectSpecImpl<>(domainType, returnType, query, mapping);
        }

        @Override
        public MappingSpec<T> matching(Criteria criteria) {
            return this.matching(Query.query(criteria));
        }

        @Override
        public MappingSpec<T> matching(Criteria criteria, Pageable pageable) {
            if (pageable != null) {
                return this.matching(Query.query(criteria).with(pageable));
            } else {
                return this.matching(Query.query(criteria));
            }
        }

        @Override
        public <R> TerminatingSpec<R> mapTo(Function<T, R> mapping) {
            return new SelectSpecImpl<>(domainType, returnType, query, mapping);
        }

        @Override
        public Mono<Page<T>> execute() {
            return this.execute(SpringContext.getBean(R2dbcEntityTemplate.class));
        }

        @SuppressWarnings("unchecked")
        @Override
        public Mono<Page<T>> execute(R2dbcEntityTemplate template) {
            Query countQuery = query.getCriteria().map(Query::query).orElse(Query.empty());

            Mono<Long> count = template.select(domainType).as(returnType).matching(countQuery).count();

            Flux<T> all = (Flux<T>) template.select(domainType).as(returnType).matching(query).all();

            if (mapping != null) {
                Function<Object, T> mapping = (Function<Object, T>) this.mapping;
                all = all.map(mapping);
            }

            return Mono.zip(all.collectList(), count).map(tp -> {
                Pageable pageable = Pageable.unpaged();
                int size = query.getLimit();
                if (size > 0) {
                    int page = (int) (query.getOffset() / size);
                    pageable = PageRequest.of(page, size, query.getSort());
                }

                return new PageImpl<>(tp.getT1(), pageable, tp.getT2());
            });
        }
    }
}
